package com.jy.oms.framework.asyn.task.executor;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.thread.ThreadUtil;
import com.google.common.base.Throwables;
import com.jy.oms.framework.asyn.task.alarm.AlarmMessage;
import com.jy.oms.framework.asyn.task.alarm.AlarmMessageFacade;
import com.jy.oms.framework.asyn.task.checker.TaskExecutorChainChecker;
import com.jy.oms.framework.asyn.task.checker.model.ExecutorCheckParam;
import com.jy.oms.framework.asyn.task.checker.model.ExecutorCheckResult;
import com.jy.oms.framework.asyn.task.constants.TaskConstants;
import com.jy.oms.framework.asyn.task.constants.TaskQueueTypeEnum;
import com.jy.oms.framework.asyn.task.constants.TaskStateEnum;
import com.jy.oms.framework.asyn.task.core.ExecutorContext;
import com.jy.oms.framework.asyn.task.exception.TaskNeverRetryException;
import com.jy.oms.framework.asyn.task.executor.bo.TaskExecuteResult;
import com.jy.oms.framework.asyn.task.executor.bo.TaskThreadControllerBo;
import com.jy.oms.framework.asyn.task.handler.ITaskHandler;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.model.*;
import com.jy.oms.framework.asyn.task.model.strategy.ExecuteStrategyObj;
import com.jy.oms.framework.asyn.task.queue.TaskQueueManagerProxy;
import com.jy.oms.framework.thread.CustomThreadPoolExecutor;
import com.jy.oms.framework.util.CompletableFutureUtil;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 任务分组执行器
 *
 * @author liuquan
 */
@Service
public class GroupTaskExecutor implements ITaskExecutor, DisposableBean {
    protected static Logger logger = LoggerFactory.getLogger(GroupTaskExecutor.class);
    @Autowired
    protected ITaskHandlerManager taskHandlerManager;
    protected Map<TaskGroup, TaskThreadControllerBo> taskExecutorMap = new ConcurrentHashMap<>();
    @Autowired
    private List<ITaskLifecycleListener> taskLifecycleListeners;

    @Override
    public void destroy() {
        taskExecutorMap.forEach((k, v) -> v.getThreadPoolExecutor().shutdown());
    }

    @Override
    public void init(TaskGroup taskGroup) {
        this.buildThreadPool(taskGroup);
    }

    /**
     * 每个任务组配置独立线程池，避免互相影响
     */
    private void buildThreadPool(TaskGroup taskGroup) {
        int corePoolSize;
        int taskGroupNum = taskHandlerManager.getExecutableTaskGroupNum();
        if (taskGroupNum == 1) {
            corePoolSize = Runtime.getRuntime().availableProcessors() * 2 + 1;
        } else {
            if (TaskConstants.DEFAULT_GROUP_CODE.equals(taskGroup.getGroup())) {
                corePoolSize = Runtime.getRuntime().availableProcessors() * 2 + 1;
            } else {
                //相同分组任务，核心线程数必须一样（启动时有校验）
                Map<TaskGroup, List<TaskHandlerWrapper>> taskGroupListMap = taskHandlerManager.loadAllExecutableTaskHandler();
                TaskHandlerWrapper taskHandlerWrapper = taskGroupListMap.get(taskGroup).get(0);
                ExecuteStrategyObj executeStrategy = taskHandlerWrapper.getExecuteStrategy();
                if (executeStrategy.getTaskGroupCoreThreadNum() != null && executeStrategy.getTaskGroupCoreThreadNum() > 0) {
                    corePoolSize = executeStrategy.getTaskGroupCoreThreadNum();
                } else {
                    if (taskGroup.getTaskQueueType() == TaskQueueTypeEnum.DB) {
                        corePoolSize = Runtime.getRuntime().availableProcessors();
                    } else {
                        corePoolSize = Runtime.getRuntime().availableProcessors() * 2 + 1;
                    }
                }
            }
        }

        String threadName = "GroupTaskExecutor_" + taskGroup.getTaskQueueType().name() + "_" + taskGroup.getGroup();
        logger.info("线程组 " + threadName + " 开启线程数：" + corePoolSize);

        ThreadPoolExecutor queueTaskExecutor = new CustomThreadPoolExecutor(
                threadName,
                corePoolSize,
                corePoolSize * 2,
                TaskConstants.DEFAULT_KEEP_ALIVE_TIME,
                TaskConstants.DEFAULT_KEEP_ALIVE_TIME_UNIT,
                new ArrayBlockingQueue<>(TaskConstants.DEFAULT_BLOCKING_QUEUE_SIZE),
                ThreadFactoryBuilder.create().setNamePrefix(threadName + "_").build(),
                new ThreadPoolExecutor.DiscardPolicy());
        taskExecutorMap.put(taskGroup, new TaskThreadControllerBo(queueTaskExecutor));
    }

    /**
     * 检查并重建线程池；
     * 如果堵塞任务过多导致核心线程一直被堵塞不能被释放，线程池新加任务将不能得到执行，这种情况下则重建线程池
     */
    private boolean checkThreadPoolAvailable(TaskGroup taskGroup) {
        TaskThreadControllerBo taskThreadControllerBo = taskExecutorMap.get(taskGroup);
        ThreadPoolExecutor threadPoolExecutor = taskThreadControllerBo.getThreadPoolExecutor();
        int corePoolSize = threadPoolExecutor.getCorePoolSize();
        int activeThreadCount = threadPoolExecutor.getActiveCount();
        boolean poolBusy;
        if (TaskQueueTypeEnum.DB == taskGroup.getTaskQueueType()) {
            poolBusy = activeThreadCount >= corePoolSize / 2;
        } else {
            //当前最大任务数量 等于 核心线程数；当核心线程都在运行时，不从redis/mem获取任务
            poolBusy = activeThreadCount == corePoolSize;
        }

        if (poolBusy) {
            if (taskThreadControllerBo.getLastCheckAvailableTime() == null) {
                taskThreadControllerBo.setLastCheckAvailableTime(System.currentTimeMillis());
                return false;
            }

            //尝试等待阻塞线程执行完毕
            if (System.currentTimeMillis() - taskThreadControllerBo.getLastCheckAvailableTime() < TaskConstants.DEFAULT_THREADPOOL_TASK_TIMEOUT_MIllS) {
                return false;
            }

            //当超时任务堵塞时间超过阈值，则销毁线程池
            threadPoolExecutor.shutdownNow();
            //重建线程池，DB队列模式下，数据不会丢失，可重新执行; redis/mem模式下数据会丢失
            this.buildThreadPool(taskGroup);

            AlarmMessage alarmMessage = AlarmMessage.create()
                    .title("线程池重建告警")
                    .add("任务组", taskGroup.getGroup())
                    .add("队列类型", taskGroup.getTaskQueueType())
                    .add("重建前堵塞线程数", activeThreadCount)
                    .add("重建前核心线程数", corePoolSize);
            logger.info(alarmMessage.toString());

            AlarmMessageFacade.getInstance().alarm(alarmMessage);
        } else {
            if (taskThreadControllerBo.getLastCheckAvailableTime() != null) {
                taskThreadControllerBo.setLastCheckAvailableTime(null);
            }
        }

        return true;
    }

    @Override
    public boolean isAbleExecute(TaskGroup taskGroup) {
        return taskGroup != null && this.checkThreadPoolAvailable(taskGroup);
    }

    @Override
    public void executeTask(TaskGroup taskGroup, List<TaskWrapper> taskList) {
        if (taskGroup == null || CollectionUtils.isEmpty(taskList)) {
            return;
        }

        CountDownLatch latch = null;
        //DB模式需要批量执行; mem/redis只要有可用线程，则可执行任务
        if (taskGroup.getTaskQueueType() == TaskQueueTypeEnum.DB) {
            latch = new CountDownLatch(taskList.size());
        }

        CountDownLatch finalLatch = latch;
        long begin = System.currentTimeMillis();
        TaskThreadControllerBo taskThreadControllerBo = taskExecutorMap.get(taskGroup);
        ThreadPoolExecutor threadPoolExecutor = taskThreadControllerBo.getThreadPoolExecutor();
        taskList.forEach(task -> CompletableFutureUtil
                .completeOnTimeout(CompletableFuture.runAsync(() -> executeTask(task), threadPoolExecutor),
                        Duration.ofSeconds(TaskConstants.DEFAULT_TASK_EXECUTE_TIMEOUT_SECONDS))
                .thenAccept(result -> {
                    onTasklifecycleListen(task, ITaskLifecycleListener.TaskLifecycleEnum.Finish);
                    if (finalLatch != null) {
                        finalLatch.countDown();
                    }
                })
                .exceptionally(throwable -> {
                    onTasklifecycleListen(task, ITaskLifecycleListener.TaskLifecycleEnum.Timeout);
                    if (finalLatch != null) {
                        finalLatch.countDown();
                    }
                    return null;
                }));

        if (finalLatch != null) {
            try {
                //保证当前批次任务执行完
                latch.await();
            } catch (InterruptedException e) {
                logger.error("interrupt error.", e);
            }

            long cost = System.currentTimeMillis() - begin;
            logger.info("当前批次任务执行完毕，taskGroup：" + taskGroup.getGroup() + ", 耗时：" + cost);
        }
    }

    private void onTasklifecycleListen(TaskWrapper taskWrapper,
                                       ITaskLifecycleListener.TaskLifecycleEnum taskLifecycleEnum) {
        for (ITaskLifecycleListener taskLifecycleListener : taskLifecycleListeners) {
            try {
                if (taskLifecycleEnum == ITaskLifecycleListener.TaskLifecycleEnum.Timeout) {
                    taskLifecycleListener.onTimeout(taskWrapper);
                } else if (taskLifecycleEnum == ITaskLifecycleListener.TaskLifecycleEnum.Finish) {
                    taskLifecycleListener.onFinish(taskWrapper);
                } else if (taskLifecycleEnum == ITaskLifecycleListener.TaskLifecycleEnum.Filter) {
                    taskLifecycleListener.onFilter(taskWrapper);
                }
            } catch (Exception e) {
                logger.error("TaskLifecycle error, taskCode: {}, billCode:{}, uuid: {}",
                        taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), e);
            }
        }
    }

    @Override
    public TaskResultDto executeTask(TaskWrapper taskWrapper) {
        TaskResultDto taskResultDto = new TaskResultDto();
        taskResultDto.setTaskWrapper(taskWrapper);
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskWrapper.getTaskCode());
        if (!taskWrapper.isExecuteImmediate()) {
            long beginTime = System.currentTimeMillis();
            ExecutorCheckParam executorCheckParam = new ExecutorCheckParam(taskWrapper.getTaskCode());
            executorCheckParam.setParamIdentifier(taskWrapper.getParamIdentifier());
            executorCheckParam.setBillCode(taskWrapper.getTaskDto().getBillCode());
            //链式校验
            ExecutorCheckResult checkResult = TaskExecutorChainChecker.getInstance().canExecute(executorCheckParam);
            taskResultDto.setCheckerCostTime(System.currentTimeMillis() - beginTime);
            if (checkResult.isResult()) {
                return this.executeTaskHandler(taskHandlerWrapper, taskWrapper, taskResultDto);
            } else {
                taskResultDto.setCheckFilter(true);
                taskResultDto.setExecutorCheckResult(checkResult);
                this.onTasklifecycleListen(taskWrapper, ITaskLifecycleListener.TaskLifecycleEnum.Filter);
                try {
                    TaskQueueManagerProxy.locate(taskHandlerWrapper.getTaskQueue()).finish(taskResultDto);
                } catch (Exception e) {
                    logger.error("finish error, taskCode: {}, billCode:{}, uuid: {}",
                            taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), e);
                }
                return taskResultDto;
            }
        } else {
            return this.executeTaskHandler(taskHandlerWrapper, taskWrapper, taskResultDto);
        }
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    private TaskResultDto executeTaskHandler(TaskHandlerWrapper taskHandlerWrapper,
                                             TaskWrapper taskWrapper,
                                             TaskResultDto taskResultDto) {
        long beginTime = System.currentTimeMillis();
        //如果任务有重试过，则判断累计执行次数是否超过限制
        taskResultDto.setRetryNumConfig(taskHandlerWrapper.getTaskConfig().getRetryNum());
        taskResultDto.setCallCount(taskWrapper.getCallCount());
        TaskExecuteResult taskExecuteResult = this.executeTaskHandlerRetry(taskHandlerWrapper, taskWrapper, taskResultDto);
        taskResultDto.setCostTime(System.currentTimeMillis() - beginTime);
        taskResultDto.setTaskResult(taskExecuteResult.getResult());

        ITaskHandler taskHandler = taskHandlerWrapper.getTaskHandler();
        long beginCallbackTime = System.currentTimeMillis();
        TaskStateEnum callbackState = TaskStateEnum.DONE;
        if (taskExecuteResult.isSuccess()) {
            if (taskExecuteResult.getResult() != null) {
                try {
                    taskHandler.executeTaskCallback(taskWrapper.getTaskDto().getParam(), taskExecuteResult.getResult());
                } catch (Throwable e) {
                    if (ExecutorContext.isBusinessException(e)) {
                        logger.warn("执行任务回调方法报错（业务异常）, taskCode: {}, billCode:{}, uuid: {}, error:{}",
                                taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), e.getMessage());
                    } else {
                        logger.error("执行任务回调方法报错, taskCode: {}, billCode:{}, uuid: {}",
                                taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), e);
                    }
                    callbackState = TaskStateEnum.ERROR;
                }
            }
        } else {
            //如果任务重试次数大于0，只有当最后一次执行报错时才会执行此回调
            if (!taskResultDto.isEnableRetry() || !taskWrapper.isExecuteImmediateRetryOnFailure()) {
                try {
                    taskHandler.executeTaskOnFailure(taskWrapper.getTaskDto().getParam());
                } catch (Throwable e) {
                    if (ExecutorContext.isBusinessException(e)) {
                        logger.warn("任务执行失败回调处理报错（业务异常）, taskCode: {}, billCode:{}, uuid: {}, error:{}",
                                taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), e.getMessage());
                    } else {
                        logger.error("任务执行失败回调处理报错, taskCode: {}, billCode:{}, uuid: {}",
                                taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), e);
                    }
                }
            }
        }

        if (taskExecuteResult.isExceedRetryMaxNum()) {
            try {
                TaskExceedRetryMaxNumDto taskExceedRetryMaxNumDto = new TaskExceedRetryMaxNumDto();
                taskExceedRetryMaxNumDto.setTaskCode(taskWrapper.getTaskCode());
                taskExceedRetryMaxNumDto.setParam(taskWrapper.getTaskDto().getParam());
                taskExceedRetryMaxNumDto.setBillNo(taskWrapper.getTaskDto().getBillCode());
                taskExceedRetryMaxNumDto.setUuid(taskWrapper.getParamIdentifier());
                taskExceedRetryMaxNumDto.setExceptionType(taskResultDto.getExceptionType());
                taskExceedRetryMaxNumDto.setExceptionTitle(taskResultDto.getExceptionTitle());
                taskHandler.exceedRetryMaxNumCallback(taskExceedRetryMaxNumDto);
            } catch (Throwable e) {
                logger.error("超过最大重试次数时回调处理报错, taskCode: {}, billCode:{}, uuid: {}",
                        taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), e);
            }
        }
        
        if(taskExecuteResult.isTaskNeverRetryAlarm()){
            AlarmMessage alarmMessage = AlarmMessage.create()
                    .add("taskCode", taskWrapper.getTaskCode())
                    .add("uuid", taskWrapper.getParamIdentifier());
            if(StringUtils.isNotBlank(taskResultDto.getExceptionTitle())){
                alarmMessage.add("exception", taskResultDto.getExceptionTitle());
            }
            AlarmMessageFacade.getInstance().alarm(alarmMessage);
        }

        taskResultDto.setCallbackState(callbackState);
        taskResultDto.setCallbackCostTime(System.currentTimeMillis() - beginCallbackTime);

        try {
            TaskQueueManagerProxy.locate(taskHandlerWrapper.getTaskQueue()).finish(taskResultDto);
        } catch (Exception e) {
            logger.error("finish error, taskCode: {}, billCode:{}, uuid: {}",
                    taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), e);
        }

        return taskResultDto;
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    private TaskExecuteResult executeTaskHandlerRetry(TaskHandlerWrapper taskHandlerWrapper,
                                                      TaskWrapper taskWrapper,
                                                      TaskResultDto taskResultDto) {
        ITaskHandler taskHandler = taskHandlerWrapper.getTaskHandler();
        Object result = null;
        boolean exceedRetryMaxNum = false;
        boolean taskNeverRetryAlarm = false;
        try {
            taskResultDto.setCallCount(taskResultDto.getCallCount() + 1);
            result = taskHandler.executeTask(taskWrapper.getTaskDto().getParam());
            taskResultDto.setTaskState(TaskStateEnum.DONE);
            return new TaskExecuteResult().setSuccess(true).setResult(result).setExceedRetryMaxNum(exceedRetryMaxNum);
        } catch (Throwable e) {
            boolean neverRetry = e instanceof TaskNeverRetryException;
            boolean executeImmediateNeverRetry = taskWrapper.isExecuteImmediate() && !taskWrapper.isExecuteImmediateRetryOnFailure();
            TaskConfig taskConfig = taskHandlerWrapper.getTaskConfig();
            int retryNum = taskConfig.getRetryNum();
            if (retryNum > 0 && taskResultDto.getCallCount() <= retryNum && !neverRetry && !executeImmediateNeverRetry) {
                if (taskConfig.isRetryContinuous()) {
                    if (taskConfig.getRetryIntervalSeconds() > 0) {
                        int sleepSeconds = taskConfig.getRetryIntervalSeconds() < TaskConstants.DEFAULT_TASK_EXECUTE_TIMEOUT_SECONDS
                                ? taskConfig.getRetryIntervalSeconds() / taskConfig.getRetryNum() : TaskConstants.DEFAULT_TASK_EXECUTE_TIMEOUT_SECONDS / taskConfig.getRetryNum();
                        ThreadUtil.sleep(sleepSeconds, TimeUnit.SECONDS);
                    }
                    return this.executeTaskHandlerRetry(taskHandlerWrapper, taskWrapper, taskResultDto);
                } else {
                    //超时任务，更新下一次最早执行时间，待下一次查询执行
                    taskResultDto.setUpdateTaskParamWhenException(taskConfig.isUpdateTaskParamWhenException());
                    taskResultDto.setEnableRetry(true);
                    taskResultDto.setTaskState(TaskStateEnum.ERROR);
                    taskResultDto.setRetryIntervalSeconds(taskConfig.getRetryIntervalSeconds());
                    taskResultDto.setRetryBackoff(taskConfig.isRetryBackoff());
                    this.fillTaskResultException(taskWrapper, taskResultDto, e);
                    return new TaskExecuteResult().setSuccess(false).setResult(result).setExceedRetryMaxNum(exceedRetryMaxNum);
                }
            } else {
                if(executeImmediateNeverRetry){
                    exceedRetryMaxNum = true;
                }else if(neverRetry){
                    TaskNeverRetryException taskNeverRetryException = (TaskNeverRetryException) e;
                    taskNeverRetryAlarm = taskNeverRetryException.isAlarm();
                }else{
                    exceedRetryMaxNum = retryNum == 0 || taskResultDto.getCallCount() > retryNum;
                }
                taskResultDto.setTaskState(TaskStateEnum.ERROR);
                this.fillTaskResultException(taskWrapper, taskResultDto, e);
            }
        }

        return new TaskExecuteResult().setSuccess(false).setResult(result).setExceedRetryMaxNum(exceedRetryMaxNum).setTaskNeverRetryAlarm(taskNeverRetryAlarm);
    }

    private void fillTaskResultException(TaskWrapper taskWrapper, TaskResultDto taskResultDto, Throwable e){
        if(ExecutorContext.isBusinessException(e)){
            taskResultDto.setExceptionType(1);
            taskResultDto.setExceptionTitle(StringUtils.defaultIfBlank(e.getMessage(), ""));
            taskResultDto.setResponse("(业务异常)" + this.getExceptionMsg(e));
        }else{
            taskResultDto.setExceptionType(2);
            taskResultDto.setExceptionTitle(StringUtils.defaultIfBlank(e.getMessage(), ""));
            taskResultDto.setResponse(this.getExceptionDetail(e));
        }

        if (ExecutorContext.isBusinessException(e)) {
            logger.warn("执行任务报错（业务异常）, taskCode: {}, billCode:{}, uuid: {}, 执行次数:{}, 业务异常:{}",
                    taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), taskResultDto.getCallCount(), e.getMessage(), e);
        } else {
            logger.error("执行任务报错, taskCode: {}, billCode:{}, uuid: {}, 执行次数:{}",
                    taskWrapper.getTaskDto().getTaskCode(), taskWrapper.getTaskDto().getBillCode(), taskWrapper.getParamIdentifier(), taskResultDto.getCallCount(), e);
        }
    }


    private String getExceptionMsg(Throwable e){
        Throwable throwable = Throwables.getRootCause(e);
        String message = throwable.getMessage();
        if(message == null){
            return null;
        }
        return message.length() > TaskConstants.TASK_FAIL_RES_MESSAGE_LIMIT ? message.substring(0, TaskConstants.TASK_FAIL_RES_MESSAGE_LIMIT) : message;
    }

    private String getExceptionDetail(Throwable e){
        Throwable throwable = Throwables.getRootCause(e);
        return ExceptionUtil.stacktraceToString(throwable, TaskConstants.TASK_FAIL_RES_MESSAGE_LIMIT);
    }
}
